package com.lpf.grpc;

import com.lpf.generated.proto.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Iterator;

/**
 * 功能描述：客户端
 *
 * @author: lpf
 * @date: 2019/6/22 12:10
 * @version: 1.0
 */
public class GrpcClient {

    public static void main(String[] args) throws IOException, InterruptedException {

        /**
         * 客户端和服务端建立的是TCP连接，建立后连接会一直保持
         */
        ManagedChannel managedChannel = ManagedChannelBuilder
                                                    .forAddress("localhost", 8899)
                                                    .usePlaintext(true)
                                                    .build();

        // 客户端用来访问服务端的stub
        // 客户端用来和服务端进行交互的真正的对象，它底层会发送相应的TCP请求

        // 同步，阻塞：等待另一端处理完成后才能往下进行
        StudentServiceGrpc.StudentServiceBlockingStub blockingStub = StudentServiceGrpc.newBlockingStub(managedChannel);

        // 异步，非阻塞：调用流程会继续往下走，不在等待
        //StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc.newStub(managedChannel);


        noStream(blockingStub);
        //serverStream(blockingStub);

        //clientStream(stub);
        //serverStreamAndClientStream(stub);
    }

    private static void noStream(StudentServiceGrpc.StudentServiceBlockingStub blockingStub) throws IOException {
        MyResponse myResponse = blockingStub.getRealNameByUserName(MyRequest.newBuilder().setUsername("zhangsan").build());
        System.out.println(myResponse.getRealname());
        System.in.read(); // 不让客户端关闭，让连接一直保持，只有这样才能复用连接，才能体现出socket编程的优势（可以通过心跳来检测连接是否关闭）
    }

    private static void serverStream(StudentServiceGrpc.StudentServiceBlockingStub blockingStub) {
        // 流式数据就是一个迭代器（流式数据反应到代码里就是一个迭代器）
        Iterator<StudentResponse> responseIterator = blockingStub.getStudentsByAge(StudentRequest.newBuilder().setAge(20).build());
        while (responseIterator.hasNext()) {
            StudentResponse studentResponse = responseIterator.next();
            System.out.println(studentResponse.getName() + ", " + studentResponse.getAge() + ", " + studentResponse.getCity());
        }
    }

    private static void clientStream(StudentServiceGrpc.StudentServiceStub stub) throws InterruptedException {

        // 服务器端响应对象
        // 服务器端给客户端返回数据之后，客户端在特定的方法上会被回调
        StreamObserver<StudentResponseList> responseStreamObserver = new StreamObserver<StudentResponseList>() {

            // 服务端向客户端返回响应结果，客户端的onNext方法会被触发
            @Override
            public void onNext(StudentResponseList value) {
                value.getStudentResponseList().forEach(studentResponse -> {
                    System.out.println(studentResponse.getName());
                    System.out.println(studentResponse.getAge());
                    System.out.println(studentResponse.getCity());
                    System.out.println("**********************");
                });
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("completed！");
            }
        };

        // 以下是客户端构造向服务端发送的数据
        StreamObserver<StudentRequest> requestStreamObserver = stub.getStudentsWrapperByAges(responseStreamObserver);

        requestStreamObserver.onNext(StudentRequest.newBuilder().setAge(20).build());
        requestStreamObserver.onNext(StudentRequest.newBuilder().setAge(30).build());
        requestStreamObserver.onNext(StudentRequest.newBuilder().setAge(40).build());
        requestStreamObserver.onNext(StudentRequest.newBuilder().setAge(50).build());

        // 表示客户端调用结束
        requestStreamObserver.onCompleted();

        Thread.sleep(5000); // 休眠一段时间，否则看不到数据。因为是异步的，所以还没等数据发出程序就退出了。
    }

    private static void serverStreamAndClientStream(StudentServiceGrpc.StudentServiceStub stub) throws InterruptedException {

        StreamObserver<StreamRequest> requestStreamObserver = stub.biTalk(new StreamObserver<StreamResponse>() {
            @Override
            public void onNext(StreamResponse value) {
                System.out.println(value.getResponseInfo());
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(t.getMessage());
            }

            @Override
            public void onCompleted() {
                // 服务器关闭了，客户端将信打印出来
                System.out.println("onCompleted!");
            }
        });

        // 为了验证时流式发送数据，这里使用循环
        for (int i = 0; i < 10; i++) {
            requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo(LocalDateTime.now().toString()).build());
            Thread.sleep(1000);
        }
    }

}
